package com.baijiaoxi.common.utils.jdk;

import com.baijiaoxi.common.model.UUID;
import com.baijiaoxi.common.model.local.ThreadLocalUtil;
import com.baijiaoxi.common.model.request.RequestInfo;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.MDC;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@Slf4j
public class StreamUtil {
    public static int localMaxExecutorSize = 8;

    /**
     * 阻塞
     *
     * @param list
     * @param streamRun
     * @param maxExecutorSize
     * @param <T>
     */
    public static <T> void forEach(List<T> list, StreamRun<T> streamRun, int maxExecutorSize) {
        int size = list.size();
        int executorSize = size > maxExecutorSize ? maxExecutorSize : size;
        ExecutorService executorService = Executors.newFixedThreadPool(executorSize);
        CountDownLatch countDownLatch = new CountDownLatch(size);
        RequestInfo requestInfo = ThreadLocalUtil.get();
        list.stream().forEach(item -> {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        MDC.put("requestId", requestInfo != null ? requestInfo.getRequestId() : UUID.uuid());
                        streamRun.run(item);
                    } catch (Exception e) {
                        log.error("streamRun.run异常，item:{}", item, e);
                        log.error("streamRun.run异常", e);
                    } finally {
                        countDownLatch.countDown();
                        MDC.put("requestId", null);
                    }
                }
            });
        });
        try {
            countDownLatch.await();
            executorService.shutdown();
        } catch (InterruptedException interruptedException) {
            log.error("countDownLatch.await", interruptedException);
        }
    }

    public static <T> void forEach(List<T> list, StreamRun<T> streamRun) {
        forEach(list, streamRun, localMaxExecutorSize);
    }

    public static <T> void forEachWithEmptyList(List<T> list, StreamRun<T> streamRun, int maxExecutorSize) {
        if (list != null && !list.isEmpty())
            forEach(list, streamRun, maxExecutorSize);
    }

    public static <T> void forEachWithEmptyList(List<T> list, StreamRun<T> streamRun) {
        forEachWithEmptyList(list, streamRun, localMaxExecutorSize);
    }

    /**
     * 无阻塞
     *
     * @param list
     * @param streamRun
     * @param <T>
     */
    public static <T> void forEachNoBlock(List<T> list, StreamRun<T> streamRun) {
        int size = list.size();
        ExecutorService executorService = Executors.newFixedThreadPool(size);
        RequestInfo requestInfo = ThreadLocalUtil.get();
        list.stream().forEach(item -> {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    try {
                        MDC.put("requestId", requestInfo != null ? requestInfo.getRequestId() : UUID.uuid());
                        streamRun.run(item);
                    } catch (Exception e) {
                        log.error("streamRun.run异常，item:{}", item, e);
                        log.error("streamRun.run异常", e);
                    } finally {
                        MDC.put("requestId", null);
                    }
                }
            });

        });
        executorService.shutdown();
    }

    public static <T> void forEachNoBlockWithEmptyList(List<T> list, StreamRun<T> streamRun) {
        if (list != null && !list.isEmpty())
            forEachNoBlock(list, streamRun);
    }
}
